package enode

import (
	"bufio"
	"bytes"
	"common/erpc"
	"context"
	"crypto/md5"
	"enode/store"
	"errors"
	"fmt"
	"github.com/google/uuid"
	"go.uber.org/zap"
	"io"
	"os"
)

type ERPCServer struct {
	log    *zap.Logger
	config *Config
	erpc.RpcPieceStoreServer
	store store.Store
}

func NewERPCServer(log *zap.Logger, cfg Config, store store.Store) *ERPCServer {
	return &ERPCServer{config: &cfg, log: log, store: store}
}

func (server *ERPCServer) Upload(requst erpc.UploadRequest, stream erpc.UploadStream) error {
	//接收数据写入临时文件
	tmpFile, err := os.CreateTemp("", "upload")
	if err != nil {
		err = fmt.Errorf("PiecestoreServer.Upload 创建临时文件失败：%s", err)
		server.log.Error(err.Error())
		return err
	}
	defer func() {
		_ = tmpFile.Close()
		_ = os.Remove(tmpFile.Name())
	}()
	bufWriter := bufio.NewWriter(tmpFile)
	total := int(0)
	//var newPieceId = string(requst.PieceId)
	md5Value := md5.Sum(requst.PieceId)
	newPieceId, err := uuid.FromBytes(md5Value[:])
	if err != nil {
		err = fmt.Errorf("uuid.FromBytes err ：%s", err)
		server.log.Error(err.Error())
		return err
	}

	server.log.Info("upload ", zap.String("", newPieceId.String()))
	for {
		message, err := stream.Recv()
		if err != nil {
			server.log.Error("stream.Recv err ", zap.Error(err))
			return err
		}
		if message == nil {
			err = errors.New("message is nil")
			server.log.Error(err.Error())
			return err
		}
		if message.Done == 1 {
			server.log.Info("message.Done")
			break
		}
		if message.Offset != int(total) {
			server.log.Error("err offset", zap.Int("total", total), zap.Int("message.Offset", message.Offset))
			return errors.New("err offset")
		}
		n, err := bufWriter.Write(message.Data)
		if err != nil {
			server.log.Error("## recv data err ", zap.Int("len", n), zap.Error(err))
			return errors.New("recv data err")
		}
		total += n
	}
	if err = bufWriter.Flush(); err != nil {
		server.log.Error("## bufWriter.Flush err ", zap.Error(err))
		return errors.New("recv data err")
	}
	err = stream.SendAndClose(&erpc.UploadResult{
		Hash: []byte("hash"),
	})

	server.log.Error("## recv upload ", zap.Int("len", total), zap.String("tmpFile", tmpFile.Name()))
	//指针移回到初始位置
	_, err = tmpFile.Seek(0, 0)
	if err != nil {
		return err
	}
	ctx := context.Background()
	err = server.store.PutPiece(ctx, newPieceId, requst.ExpireTime, total, tmpFile, [16]byte{})
	server.log.Info(fmt.Sprintf("%s写入：%s, size: %d", server.store.Mode(), newPieceId, total))

	server.log.Info("上传已完成", zap.Error(err))
	return err
}

func (server *ERPCServer) Download(requst erpc.DownloadRequst, stream erpc.DownloadStream) error {

	md5Value := md5.Sum(requst.PieceId)
	pieceId, err := uuid.FromBytes(md5Value[:])
	if err != nil {
		err = fmt.Errorf("uuid.FromBytes err ：%s", err)
		server.log.Error(err.Error())
		return err
	}
	ctx := context.Background()
	data, err := server.store.GetPiece(ctx, pieceId)
	server.log.Info(fmt.Sprintf("读取：%s,  size: %d", pieceId.String(), len(data)), zap.Error(err))
	if err != nil {
		return err
	}
	reader := bytes.NewReader(data)

	var maximumStep = 1024
	//done := false
	var offset int = 0
	sendbuf := make([]byte, maximumStep)
	for {
		var done = 0
		n, errRead := reader.Read(sendbuf)
		if errRead == io.EOF {
			done = 1
		}
		sendErr := stream.Send(&erpc.Chunk{
			Offset: offset,
			Data:   sendbuf[:n],
			Done:   done,
		})
		if sendErr != nil {
			server.log.Debug("stream.Send err", zap.Error(sendErr))
			err = sendErr
			break
		}
		offset += n
		if errRead != nil {
			server.log.Info("errRead break", zap.Int("offset", offset), zap.Error(errRead))
			err = errRead
			break
		}
	}
	return err
}

func (server *ERPCServer) Run(ctx context.Context) error {
	return erpc.StartRpcServer(ctx, server.config.Address, server, true)
}
